package com.xunwulian.cmpp.server.handler;

import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.filter.executor.OrderedThreadPoolExecutor;
import org.apache.mina.integration.jmx.IoServiceMBean;
import org.apache.mina.transport.socket.SocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetSocketAddress;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

@Component
public class MinaCmpp extends IoHandlerAdapter implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(MinaCmpp.class);

    private static final int PORT = 7890;
    private static final int BUFFER_SIZE = 81920;

    @Autowired
    private  CmppServerIoHandler ioHandler;

    public MinaCmpp() {
    }

    /**
     * run mina server and return acceptor
      */
    public SocketAcceptor runMinaServer() {

        SocketAcceptor socketAcceptor = new NioSocketAcceptor(Runtime.getRuntime().availableProcessors() + 1);
        socketAcceptor.setReuseAddress(true);
        socketAcceptor.getSessionConfig().setReceiveBufferSize(BUFFER_SIZE);

        ThreadFactory threadFactory = new ThreadFactory() {
            public Thread newThread(final Runnable r) {
                return new Thread(null, r, "MinaThread", 64 * 1024 * 1024);
            }
        };
        OrderedThreadPoolExecutor executor = new OrderedThreadPoolExecutor(0,1000,
                60, TimeUnit.SECONDS, threadFactory);
        socketAcceptor.getFilterChain().addLast("threadPool", new ExecutorFilter(executor));
        socketAcceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new CmppProtocolCodecFactory()));

        MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
        IoServiceMBean acceptorMBean = new IoServiceMBean(socketAcceptor);
        ObjectName acceptorName;

        try {
            acceptorName = new ObjectName(socketAcceptor.getClass().getPackage().getName() + ":type=acceptor,name="
                    + socketAcceptor.getClass().getSimpleName());
            mBeanServer.registerMBean(acceptorMBean, acceptorName);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return socketAcceptor;
    }

    @Override
    public void exceptionCaught(IoSession session, Throwable cause) {
        if (!(cause instanceof IOException)) {
            logger.error("Exception: ", cause);
        } else {
            logger.info("I/O error: " + cause.getMessage());
        }
        logger.info("server close");
        session.close(true);
    }

    @Override
    public void run() {
        try {
            SocketAcceptor acceptor = runMinaServer();
            final InetSocketAddress socketAddress = new InetSocketAddress("127.0.0.1", PORT);
            acceptor.setHandler(ioHandler);
            acceptor.bind(socketAddress);
            logger.info("Cmpp Mina server started，listening on：" + PORT);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String args[]) {
//        ApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring/ApplicationContext-base.xml");
        MinaCmpp server = new MinaCmpp();
        server.run();
    }

}
